package com.coffee.test;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import reactor.util.function.Tuple2;

import java.time.Duration;

/**
 * @author songkui
 * @since 2024/8/12 13:57
 */
public class T {
    public static void main(String[] args) throws InterruptedException {
        flatMap();

    }

    private static void flatMap() throws InterruptedException {
        // 使⽤flatMap()⽅法和subscribeOn()⽅法
        Flux<Integer> integerFlux = Flux
                .just("Michael", "Scottie Pippen", "Steve Kerr Ob")
                .flatMap(n -> Mono.just(n)
                        .map(p -> {
                            System.out.println("--------nn------------");
                            String[] split = p.split("\\s");
                            return split.length; // 将String转为Integer
                        })
                        .subscribeOn(Schedulers.parallel()) // 定义异步
                );
        integerFlux.subscribe(System.out::println);
// 阻塞，等待结果
        Thread.sleep(100000);


    }

    private static void map(){
        // map将元素映射为新的元素，并创建一个新的Flux
        // 其中重要的⼀点是：在每个数据项被源Flux发布时，map操作是同步执⾏的，如果你想要异步地转换过程，那么你应该考虑使⽤flatMap操作
        Flux<Integer> integerFlux = Flux
                .just("Michael Jordan", "Scottie Pippen", "Steve Kerr")
                .map(n -> {
                    System.out.println("--------nn------------");
                    String[] split = n.split("\\s");
                    return split.length; // 将String转为Integer
                });
        integerFlux.subscribe(System.out::println);
        System.out.println("--------222------------");

/**
 * 执行结果：
 * 2
 * 2
 * 2
 */

    }


    private static void skip(){
        // 跳过3个，并创建一个新的Flux
        Flux<String> skipFlux = Flux.just(
                        "one", "two", "skip a few", "ninety nine", "one hundred")
                .skip(3);
        skipFlux.subscribe(System.out::println);  //"ninety nine", "one hundred"

    }

    private static void testZip(){
        Flux<String> characterFlux = Flux
                .just("Garfield", "Kojak", "Barbossa");
        Flux<String> foodFlux = Flux
                .just("Lasagna", "Lollipops", "Apples");
// 当两个Flux对象压缩在⼀起的时候，它将会产⽣⼀个新的发布元组的Flux，其中每个元组中都包含了来⾃每个源Flux的数据项
// 这个合并后的Flux发出的每个条⽬都是⼀个Tuple2（⼀个容纳两个其他对象的容器对象）的实例，其中包含了来⾃每个源Flux的数据项，并保持着它们发布的顺序。
        Flux<Tuple2<String, String>> zippedFlux =
                Flux.zip(characterFlux, foodFlux);

        zippedFlux.subscribe(t -> {
            System.out.println(t.getT1() + "|" + t.getT2());
        });

    }

    private static void first() throws InterruptedException {
        Flux<String> slowFlux = Flux.just("tortoise", "snail", "sloth")
                .delaySubscription(Duration.ofMillis(100)); // 延迟100ms
        Flux<String> fastFlux = Flux.just("hare", "cheetah", "squirrel", "squirrel22", "squirrel33") .delaySubscription(Duration.ofMillis(201));
// 选择第⼀个反应式类型进⾏发布
        Flux<String> firstFlux = Flux.firstWithSignal(slowFlux, fastFlux);
        firstFlux.subscribe(System.out::println);
// 阻塞，等待结果
        Thread.sleep(100000);
    }


    private static void delay() throws InterruptedException {
        long s = System.currentTimeMillis();
        Flux<String> characterFlux = Flux
                .just("Garfield", "Kojak", "Barbossa")
                .delayElements(Duration.ofMillis(5000)); // 每500毫秒发布⼀个数据

        System.out.println("-------11--------"+ (System.currentTimeMillis() - s) );
        s = System.currentTimeMillis();

        Flux<String> foodFlux = Flux
                .just("Lasagna", "Lollipops", "Apples")
                .delaySubscription(Duration.ofMillis(2500)) // 订阅后250毫秒后开始发布数据
                .delayElements(Duration.ofMillis(5000)); // 每500毫秒发布⼀个数据
        System.out.println("----------22-----"+ (System.currentTimeMillis() - s) );
        s = System.currentTimeMillis();
// 使⽤mergeWith()⽅法，将两个Flux合并，合并过后的Flux数据项发布顺序与源Flux的发布时间⼀致
// Garfield Lasagna Kojak Lollipops Barbossa Apples
        Flux<String> mergedFlux = characterFlux.mergeWith(foodFlux);
        mergedFlux.subscribe(System.out::println);
        System.out.println("---------33------"+ (System.currentTimeMillis() - s) );
// 阻塞，等待结果
        Thread.sleep(100000);
    }
}
